package belf.migrate.engine.job.verify;

import belf.migrate.api.exception.CatalogException;
import belf.migrate.api.table.schema.AbstractJdbcCatalog;
import belf.migrate.api.table.schema.AbstractSinkCatalog;
import belf.migrate.api.table.schema.IndexInfo;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.type.DbObjectType;
import belf.migrate.api.type.LogLevelType;
import belf.migrate.api.util.DatabaseType;
import belf.migrate.api.util.PostLog;
import belf.migrate.core.number.NumberFormatUtil;
import belf.migrate.api.job.Job;
import belf.migrate.engine.job.model.verify.VerifyBaseModel;
import belf.migrate.engine.job.model.verify.VerifyIndexResult;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.stream.Collectors;

import static belf.migrate.api.job.JobCode.*;

@Slf4j
public class VerifyIndexTool {

    private Job job;

    private JobContext jobContext;

    private AbstractJdbcCatalog sourceCatalog;

    private AbstractSinkCatalog sinkCatalog;

    public VerifyIndexTool(JobContext jobContext, Job job) {
        this.jobContext = jobContext;
        this.sourceCatalog = jobContext.getSourceCatalog();
        this.sinkCatalog = jobContext.getSinkCatalog();
        this.job = job;
    }

    public VerifyIndexResult run() {

        DatabaseType sourceDatabaseType = DatabaseType.getDatabaseType(job.getTask().getTaskConf().getSourceConf().getCatalogName());
        DatabaseType sinkDatabaseType = DatabaseType.getDatabaseType(job.getTask().getTaskConf().getSinkConf().getCatalogName());
        if (!(sourceDatabaseType != null &&
                sinkDatabaseType != null &&
                sourceDatabaseType.getObjectTypes().contains(DbObjectType.INDEX) &&
                sinkDatabaseType.getObjectTypes().contains(DbObjectType.INDEX))) {
            log.info("当前数据库不支持索引校验!");
            return null;
        }

        log.info("开始进行索引校验!");
        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "开始进行索引校验!",
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());

        VerifyIndexResult indexResult = new VerifyIndexResult();
        //获取数据库信息
        String sourceDatabaseName = job.getTask().getTaskConf().getSourceConf().getDatabaseName();
        String sinkDatabaseName = job.getTask().getTaskConf().getSinkConf().getDatabaseName();
        String sourceSchemaName = job.getTask().getTaskConf().getSourceConf().getSchemaName();
        String sinkSchemaName = job.getTask().getTaskConf().getSinkConf().getSchemaName();
        try {
            //获取源库和目标库的所有索引
            List<IndexInfo> sourceIndexNameList = sourceCatalog.listIndexInfos(sourceDatabaseName, sourceSchemaName);
            List<IndexInfo> sinkIndexNameList = sinkCatalog.listIndexInfos(sinkDatabaseName, sinkSchemaName);
            //判断源库索引是否为空
            if (sourceIndexNameList == null) {
                if (sinkIndexNameList == null) {
                    log.info("源数据库和目标数据库索引均为空!");
                    PostLog.sendLogModel(
                            job.getJobId(),
                            job.getTask().getTaskType().getName(),
                            "源数据库和目标数据库索引均为空!",
                            LogLevelType.INFO.getType(),
                            EXECUTING,
                            jobContext.getProgress());
                    indexResult.setCode(SUCCESS);
                    indexResult.setAllSize(0);
                    indexResult.setSuccessCount(0);
                    indexResult.setFailCount(0);
                    indexResult.setSuccessPercent("100.0%");
                    indexResult.setFailPercent("0.0%");
                    indexResult.setSuccess(true);
                } else {
                    log.info("源数据库索引为空，目标数据库索引不为空!");
                    PostLog.sendLogModel(
                            job.getJobId(),
                            job.getTask().getTaskType().getName(),
                            "源数据库索引为空，目标数据库索引不为空!",
                            LogLevelType.ERROR.getType(),
                            EXECUTING,
                            jobContext.getProgress());
                    indexResult.setCode(FAILTURE);
                    indexResult.setAllSize(0);
                    indexResult.setSuccessCount(0);
                    indexResult.setFailCount(0);
                    indexResult.setSuccessPercent("0.0%");
                    indexResult.setFailPercent("100.0%");
                    indexResult.setSuccess(false);
                }

            } else {
                int allSize = sourceIndexNameList.size();
                int successCount = 0;
                int failCount = 0;
                Map<String, IndexInfo> sourceLowerIndexNameMap = sourceIndexNameList.stream().collect(
                        Collectors.toMap(s -> s.getTableName().toLowerCase(Locale.ROOT) + "." + s.getIndexName().toLowerCase(Locale.ROOT),
                                s -> s, (existing, replacement) -> existing));
                Map<String, IndexInfo> sinkLowerIndexNameMap = null;
                if (sinkIndexNameList != null) {
                    sinkLowerIndexNameMap = sinkIndexNameList.stream().collect(
                            Collectors.toMap(s -> s.getTableName().toLowerCase(Locale.ROOT) + "." + s.getIndexName().toLowerCase(Locale.ROOT),
                                    s -> s, (existing, replacement) -> existing));
                }
                List<VerifyBaseModel> indexBaseModelList = new ArrayList<>();
                int index = 0;
                //循环遍历源库索引
                for (String sourceIndexName : sourceLowerIndexNameMap.keySet()) {
                    long startTime = System.currentTimeMillis();
                    VerifyBaseModel indexBaseModel = new VerifyBaseModel();
                    if (sinkLowerIndexNameMap != null && sinkLowerIndexNameMap.containsKey(sourceIndexName)) {
                        indexBaseModel.setSuccess(true);
                        indexBaseModel.setSourceName(sourceLowerIndexNameMap.get(sourceIndexName).getIndexName());
                        indexBaseModel.setSinkName(sinkLowerIndexNameMap.get(sourceIndexName).getIndexName());
                        successCount++;
                    } else {
                        indexBaseModel.setSuccess(false);
                        indexBaseModel.setSourceName(sourceLowerIndexNameMap.get(sourceIndexName).getIndexName());
                        failCount++;
                    }
                    //计算耗时
                    long endTime = System.currentTimeMillis();
                    String timeConsuming = String.valueOf((double) (endTime - startTime) / 1000);
                    indexBaseModel.setTimeConsuming(String.format("%s秒", timeConsuming));
                    indexBaseModelList.add(indexBaseModel);

                    index++;
                    PostLog.sendLogModel(
                            job.getJobId(),
                            job.getTask().getTaskType().getName(),
                            String.format("索引 %s 校验完成，目前进度：%s / %s.", sourceIndexName, index, allSize),
                            LogLevelType.INFO.getType(),
                            EXECUTING,
                            jobContext.getProgress());
                }
                indexResult.setSuccess(successCount == allSize);
                indexResult.setAllSize(allSize);
                indexResult.setSuccessCount(successCount);
                indexResult.setFailCount(failCount);
                float successPercent = allSize == 0 ? 1 : (float)successCount / allSize;
                float failPercent = allSize == 0 ? 0 : (float)failCount / allSize;
                indexResult.setSuccessPercent(NumberFormatUtil.numberToString(successPercent, "0.0%"));
                indexResult.setFailPercent(NumberFormatUtil.numberToString(failPercent, "0.0%"));
                indexResult.setObjectList(indexBaseModelList);

            }

        } catch (CatalogException e) {
            log.error("索引校验失败！", e);
            PostLog.sendLogModel(
                    job.getJobId(),
                    job.getTask().getTaskType().getName(),
                    String.format("索引校验失败: %s", e.getMessage()),
                    LogLevelType.ERROR.getType(),
                    EXECUTING,
                    jobContext.getProgress());
            indexResult.setCode(FAILTURE);
        }
        return indexResult;
    }

}
